package com.h.consumer;

import com.h.publisher.Publisher;
import com.h.util.RabbitMQConnectionUtil;
import com.rabbitmq.client.*;
import org.junit.Test;

import java.io.IOException;
import java.io.UnsupportedEncodingException;

/**
 * @author: hlj
 * @date: 2022年05月30日 15:55
 * mq消费者
 */
public class Consumer {

    @Test
    public void Consumer() throws Exception {
        //获取链接
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //创建channel管道
        Channel channel = connection.createChannel();
        //创建队列 第一个参数队列名称 第二个是否持久化 第三个 是否排外 如果为TRUE 只允许一个消费者消费 第四个 当长时间没有被使用的话是否需要删除 第五个其他参数传map 可在可视化页面看到
        channel.queueDeclare(Publisher.QUEUE_NAME, false, false, false, null);
        //创建回调函数
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws UnsupportedEncodingException {
                System.out.println("消费者消息:" + new String(body, "UTF-8"));
            }
        };
        //监听队列 第一个 队列名 第二个ACK 拿到消息之后会给mq一个ack告诉mq已经拿到了true自动开启 false关闭需要手动告诉mq 第三个拿到消息之后会执行方法
        channel.basicConsume(Publisher.QUEUE_NAME, false, defaultConsumer);
        System.out.println("开始监听");
        System.in.read();
    }

    //WorkQueues模式
    @Test
    public void WorkQueuesConsumer1() throws Exception {
        //获取链接
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //创建channel管道
        final Channel channel = connection.createChannel();
        //一次性拿几个消息
        channel.basicQos(1);
        //创建队列 第一个参数队列名称 第二个是否持久化 第三个 是否排外 如果为TRUE 只允许一个消费者消费 第四个 当长时间没有被使用的话是否需要删除 第五个其他参数传map 可在可视化页面看到
        channel.queueDeclare(Publisher.WORK_QUEUE_NAME, false, false, false, null);
        //创建回调函数
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("消费者1消费消息:" + new String(body, "UTF-8"));
                //手动给mq ack标识 第一个参数 唯一标识 从参数里获取 第二个是不是批量操作
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        //监听队列 第一个 队列名 第二个ACK 拿到消息之后会给mq一个ack告诉mq已经拿到了true自动开启 false关闭需要手动告诉mq 第三个拿到消息之后会执行方法
        channel.basicConsume(Publisher.WORK_QUEUE_NAME, false, defaultConsumer);
        System.out.println("开始监听");
        System.in.read();
    }

    //WorkQueues模式
    @Test
    public void WorkQueuesConsumer2() throws Exception {
        //获取链接
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //创建channel管道
        final Channel channel = connection.createChannel();
        //创建队列 第一个参数队列名称 第二个是否持久化 第三个 是否排外 如果为TRUE 只允许一个消费者消费 第四个 当长时间没有被使用的话是否需要删除 第五个其他参数传map 可在可视化页面看到
        channel.queueDeclare(Publisher.WORK_QUEUE_NAME, false, false, false, null);
        //一次性拿几个消息
        channel.basicQos(1);
        //创建回调函数
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("消费者2消费消息:" + new String(body, "UTF-8"));
                //手动给mq ack标识 第一个参数 唯一标识 从参数里获取 第二个是不是批量操作
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        //监听队列 第一个 队列名 第二个ACK 拿到消息之后会给mq一个ack告诉mq已经拿到了true自动开启 false关闭需要手动告诉mq 第三个拿到消息之后会执行方法
        channel.basicConsume(Publisher.WORK_QUEUE_NAME, false, defaultConsumer);
        System.out.println("开始监听");
        System.in.read();
    }
    //RPC模式
    @Test
    public void RpcConsumer() throws Exception {
        //获取链接
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //创建channel管道
        final Channel channel = connection.createChannel();
        //创建队列 第一个参数队列名称 第二个是否持久化 第三个 是否排外 如果为TRUE 只允许一个消费者消费 第四个 当长时间没有被使用的话是否需要删除 第五个其他参数传map 可在可视化页面看到
        channel.queueDeclare(Publisher.RPC_QUEUE_PUBLISHER, false, false, false, null);
        channel.queueDeclare(Publisher.RPC_QUEUE_CONSUMER, false, false, false, null);
        //创建回调函数
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {

                System.out.println("消费者消息:" + new String(body, "UTF-8"));
                String replyTo = properties.getReplyTo();
                String correlationId = properties.getCorrelationId();
                AMQP.BasicProperties prop = new AMQP.BasicProperties().builder()
                        .correlationId(correlationId)
                        .build();
                String msg = "服务器端已接收到消息";
                //交换机 队列名称 AMQP.BasicProperties 属性
                channel.basicPublish("",replyTo,prop,msg.getBytes());
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        //监听队列 第一个 队列名 第二个ACK 拿到消息之后会给mq一个ack告诉mq已经拿到了true自动开启 false关闭需要手动告诉mq 第三个拿到消息之后会执行方法
        channel.basicConsume(Publisher.RPC_QUEUE_PUBLISHER, false, defaultConsumer);
        System.out.println("开始监听");
        System.in.read();
    }
}
